package com.bossien.flink.handler.exam.handler;

import com.bossien.flink.config.ExecutionTableConfig;
import com.bossien.flink.mapper.ExamMapper;
import com.bossien.flink.redis.FlinkRedisMapper;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.table.api.StreamQueryConfig;
import org.apache.flink.table.api.Table;
import org.apache.flink.types.Row;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;



/**
 * @author 胡春林
 * 使用sql方式查询统计
 */
@Accessors(chain = true)
@Slf4j
@Component
public class ExamStreamBySqlExecute extends ExecutionTableConfig<Table> {

    @Autowired(required = false)
    ExamMapper examMapper;

    @Override
    public Table executeFilter(Table table) {
      Table selectListBySql = examMapper.selectListBySql(table);
      if(null == selectListBySql)
      {
          log.error("返回结果为空,调用类：{}",this.getClass().getName());
          return null;
      }
        return selectListBySql;
    }

    @Override
    public Table addSkin(Table tTable) {

        //追加记录并维持状态值
//        DataStream<Tuple2<Boolean, Row>>  dataStream = getTableEnvironment().toRetractStream(tTable,Row.class);
//        Table tables =  getTableEnvironment().fromDataStream(dataStream,"UserId,Score");
//        TableSink<Row> windowSink = new CsvTableSink("E:\\ceshi\\ceshi2.csv", ",", 1, FileSystem.WriteMode.OVERWRITE);
//        TypeInformation[] fieldTypes = {Types.STRING,Types.INT};
//        String[] fieldNames = {"UserId","Score"};
//        getTableEnvironment().registerTableSink("Question",fieldNames,fieldTypes,windowSink);
//        tables.writeToSink(windowSink);
        //追加销毁时长限制
//        TableSink<Row> windowSink = new CsvTableSink("E:\\ceshi\\ceshi3.csv", ",", 1, FileSystem.WriteMode.OVERWRITE);
//        TypeInformation[] fieldTypes = {Types.STRING,Types.INT};
//        String[] fieldNames = {"UserId","Score"};
//        getTableEnvironment().registerTableSink("Question",fieldNames,fieldTypes,windowSink);
//        //限制存储非活跃数据销毁时长
        StreamQueryConfig queryConfig = getTableEnvironment().queryConfig();
        queryConfig.withIdleStateRetentionTime(Time.minutes(5),Time.minutes(10));
        //转换成dataStream
        DataStream<Tuple2<Boolean, Row>>  dataStream = getTableEnvironment().toRetractStream(tTable,Row.class,queryConfig);
       // Table tables = getTableEnvironment().fromDataStream(dataStream,"UserId,Score");
//        tables.writeToSink(windowSink);
        //数据入库到redis
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("119.23.228.209").setPort(6380).setTimeout(5000).setPassword("Bossien832192").setDatabase(13).build();
        FlinkRedisMapper flinkRedisMapper = new FlinkRedisMapper();
        flinkRedisMapper.setKey("totalScore");
        dataStream.addSink(new RedisSink(conf, flinkRedisMapper));

        return tTable;
    }
}
